package com.myownthemepark.s3vn;
import com.myownthemepark.s3vn.S3x;

import java.io.File;
import java.io.FileReader;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.FileWriter;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.lang.Integer;
import java.lang.Long;
import java.util.List;

import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.log4j.Logger;
import org.apache.log4j.BasicConfigurator;

public class S3vn implements S3vn_Interface {
	String localRepo;
	String S3Bucket;
	String S3Prefix;
	String ZKAddr;
	String ZKPrefix;
	String progName = "new";
	ZooKeeper ZKSession = null;
	S3vnStatCallback statCallback = new S3vnStatCallback();
	S3vnDataCallback dataCallback = new S3vnDataCallback();
	Watcher ZKWatcher = new S3vnWatcher();
	static Logger logger = Logger.getLogger(S3vn.class);	

	public S3vn (String localRepo, String S3Bucket, String S3Prefix, String ZKAddr, String ZKPrefix){
		this.localRepo = localRepo;
		this.S3Bucket = S3Bucket;
		this.S3Prefix = S3Prefix;
		this.ZKAddr = ZKAddr;
		this.ZKPrefix = ZKPrefix;
		//BasicConfigurator.configure();
	}
	/**
	 * Acquire a lock file in ZooKeeper. If no connection to ZooKeeper exists, restore connection.
	 * 
	 * @return
	 * @throws Exception
	 */
	public synchronized boolean acquireLock() throws Exception{
		logger.debug("acquireLock");
		String lockPath = _lock_path();
		String actualPath = "";
		String nextLowestPath = "";
		String lockNode = _lock_node();
		String lockFmtString = "%s" + S3vn.LOCK_FORMAT_STR;
		List<String> children = null;
		int lockValue;
		int childValue;
		int lowestChildValue;
		int nextLowestChildValue;
		int i;
		byte data[] = {' '};
		boolean ret = false;
		try{
			if (this.ZKSession == null) restoreZKSession();
			actualPath = this.ZKSession.create(lockPath, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
			lockValue = _convert_path_to_int(actualPath);
			
			while(true){
				try{
					children = this.ZKSession.getChildren(lockNode, null);
				}catch (KeeperException e){
					throw new Exception("Could not get children of ZooKeeper lock node: " + e.toString());
				}
				lowestChildValue = lockValue;
				nextLowestChildValue = -1;
				for (i = 0; i < children.size(); i++){
					childValue = _convert_path_to_int(children.get(i));
					if (childValue < lowestChildValue)
						lowestChildValue = childValue;
					if (childValue > nextLowestChildValue && childValue < lockValue)
						nextLowestChildValue = childValue;
				}
				// Now we can acquire the lock
				if (lowestChildValue == lockValue){
					_save_lock(lockValue);
					break;
				}
				// Otherwise we wait
				nextLowestPath = String.format(lockFmtString, _lock_path(), nextLowestChildValue);
				this.ZKSession.exists(nextLowestPath, true, this.statCallback, null);
				this.statCallback.wait();
			}
		}catch (KeeperException e){
			throw new Exception("Could not create ZooKeeper lock: " + e.toString());
		}catch (NumberFormatException e){
			throw new Exception("Malformed lock value integer for actual path: " + actualPath);
		}
		return ret;
	}
	/**
	 * Check to see if a valid lock file is found in ZooKeeper.
	 * 
	 * @return True or false depending upon whether a valid lock file is found
	 * @throws Exception
	 */
	public boolean checkLock() throws Exception{
		logger.debug("checkLock");
		long lockValue = _load_lock();
		String lockFmtString = "%s" + S3vn.LOCK_FORMAT_STR;
		String lockPath = String.format(lockFmtString, _lock_path(), lockValue);
		return this.ZKSession.exists(lockPath, false) == null ? false : true;
	}
	/**
	 * Create local Subversion repository. Pull in start-commit and post-commit hooks from S3.
	 * 
	 * @return True or False depending upon success of repository creation.
	 * @throws Exception
	 */
	public boolean createLocalRepo() throws Exception{
		logger.debug("createLocalRepo");
		boolean ret = false;
		try{
			Process process = Runtime.getRuntime().exec(String.format("%s %s", S3vn.SVNADMIN_CMD, this.localRepo));
			process.waitFor();
			switch(process.exitValue()){
				case 0: ret = true; break;
				case -1: throw new Exception("Could not execute svnadmin command");
				default: throw new Exception("svnadmin failed and returned value " + process.exitValue());
			}
			process.destroy();
			
			S3x.get(System.getenv("BUCKET_NAME"), "s3vn-start-commit.sh", String.format("%s/hooks/start-commit", this.localRepo));
			process = Runtime.getRuntime().exec(String.format("chmod 755 %s/hooks/start-commit", this.localRepo));
			process.waitFor();
			switch(process.exitValue()){
				case 0: ret = true; break;
				default: throw new Exception("Could not get s3vn-start-commit.sh from S3");
			}
			process.destroy();
			
			S3x.get(System.getenv("BUCKET_NAME"), "s3vn-post-commit.sh", String.format("%s/hooks/post-commit", this.localRepo));
			process = Runtime.getRuntime().exec(String.format("chmod 755 %s/hooks/post-commit", this.localRepo));
			process.waitFor();
			switch(process.exitValue()){
				case 0: ret = true; break;
				default: throw new Exception("Could not get s3vn-post-commit.sh from S3"); 
			}
			process.destroy();
		}catch (IOException e){
			ret = false;
			throw new Exception("[System Error] : " + e.toString());
		}
		catch (IllegalThreadStateException e){
			ret = false;
			throw new Exception("svnadmin command likely has not completed: " + e.toString());
		}
		catch (Exception e){
			ret = false;
			throw new Exception("Could not create local repository: " + e.toString());
		}
		return ret;
	}
	/**
	 * Store the repository within ZooKeeper. Store the current repository revision number in local file.
	 * 
	 * @return True or False depending upon the success of the storage of the repository.
	 * @throws Exception
	 */
	public boolean storeRepo() throws Exception{
		logger.debug("storeRepo");
		boolean ret = false;
		long current;
		int i;
		String uuid;
		uuid = getUUID();
		current = getCurrent();
		try{
			try{
				System.out.println("storeRepo: putting key: " + _uuid_key());
				S3x.put(this.S3Bucket, _uuid_key(), _uuid_filename());
			}catch (Exception e){
				throw new Exception("Could not put uuid file into S3: " + _uuid_filename());
			}
			for (i = 0; i <= current; i++){
				if (!storeRev(i)){
					throw new Exception("Could not store revision: " + i);
				}
			}
			if (this.ZKSession.exists(this.ZKPrefix, false) == null){
				this.ZKSession.create(this.ZKPrefix, uuid.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.fromFlag(0));
			}else {
				this.ZKSession.setData(this.ZKPrefix, uuid.getBytes(), -1);
			}
			if (this.ZKSession.create(_lock_node(), null, Ids.OPEN_ACL_UNSAFE, CreateMode.fromFlag(0)) == null){
				throw new Exception("Could not create ZooKeeper lock node: " + _lock_node());
			}
			storeCurrent(current);
		}catch (Exception e){
			throw new Exception("Unable to store repo: " + e.toString());
		}
		return ret;
	}
	/**
	 * Fetch repository from S3 to local storage.
	 * 
	 * @return True or False depending upon success of fetch of repository from S3.
	 * @throws Exception
	 */
	public boolean fetchRepo() throws Exception{
		logger.debug("fetchRepo");
		int currentRev = 0;
		int i = 0;
		boolean ret = false;
		
		try{
			currentRev = fetchCurrent();
			ret = setCurrent(currentRev);
			S3x.get(this.S3Bucket, _uuid_key(), _uuid_filename());
			for (i = 0; i <= currentRev; i++)
				if (!fetchRev(i))
					throw new Exception("fetchRepo: Unable to fetch revision: " + i);
		}catch (Exception e){
			ret = false;
			throw new Exception(e.toString());
		}
		return ret;
	}
	/**
	 * Store a revision in S3.
	 * 
	 * @param revNum Revision number to store in S3.
	 * @return True or False depending upon the success of the storage of the revision in S3.
	 */
	public boolean storeRev(long revNum){
		logger.debug("storeRev");
		String path = _rev_filename(revNum);
		String pathBasename = new File(path).getName();
		String propPath = _revprop_filename(revNum);
		String propPathBasename = new File(propPath).getName();
		String S3Key = _rev_key(revNum);
		String S3PropKey = _revprop_key(revNum);
		boolean ret = false;
		
		System.out.println("storeRev");
		/* Store revision data into S3 */
		while(!ret){
			try{
				System.out.println(String.format("storeRev: bucket=%s: %s: %s: %s: %s", this.S3Bucket, S3Key, path, S3PropKey, propPath));
				S3x.put(this.S3Bucket, S3Key, path);
				S3x.rename(this.S3Bucket, pathBasename, S3Key);
				S3x.put(this.S3Bucket, S3PropKey, propPath);
				S3x.rename(this.S3Bucket, propPathBasename, S3PropKey);
				ret = true;
			}
			catch (InterruptedException e){
				System.err.println("[Error]: Interrupted putting revision into S3: " + revNum);
			}catch (Exception e){
				System.err.println("[Error]: Could not put revision %d into S3: " + revNum + ": " + e.toString());
			}
		}
		return ret;
	}
	/**
	 * Fetch revision to local storage from S3.
	 * 
	 * @param revNum Revision number to fetch.
	 * @return True or False depending upon the success of the fetch.
	 * @throws Exception
	 */
	public boolean fetchRev(long revNum) throws Exception{
		String S3Key = _rev_key(revNum);
		String S3KeyProp = _revprop_key(revNum);
		String revFilename = _rev_filename(revNum);
		String revFilenameTmp = _rev_filename_tmp(revNum);
		String revPropFilename = _revprop_filename(revNum);
		String revPropFilenameTmp = _revprop_filename_tmp(revNum);
		String shardDir = _rev_shard_dir(revNum);
		String revPropShardDir = _revprop_shard_dir(revNum);
		String chmod = "chmod 0755 ";
		File newDir = null;
		File newDirRevProp = null;
		Process p = null;
		boolean ret = false;
		try{
			// Create directory and set perms
			newDir = new File(shardDir);
			try{
				newDir.mkdir();
			}catch (Exception e){}
			p = Runtime.getRuntime().exec(chmod + shardDir);
			p.waitFor();
			if (p.exitValue() != 0){ // chmod will fail if directory creation failed
				throw new IOException("Could not create chmod new directory: " + shardDir);
			}
			p.destroy();
			// End create directory...
			
			// Download revision
			S3x.get(this.S3Bucket, S3Key, revFilenameTmp);
			
			// Create directory and set perms
			newDirRevProp = new File(revPropShardDir);
			try{
				newDirRevProp.mkdir();
			}catch (Exception e){}
			p = Runtime.getRuntime().exec(chmod + revPropShardDir);
			p.waitFor();
			if (p.exitValue() != 0){ // chmod will fail if directory creation failed
				throw new IOException("Could not create or chmod new directory: " + revPropShardDir);
			}
			p.destroy();
			// End create directory...
			
			// Download revision property
			S3x.get(this.S3Bucket, S3KeyProp, revPropFilenameTmp);
			
			new File(revFilenameTmp).renameTo(new File(revFilename));
			new File(revPropFilenameTmp).renameTo(new File(revPropFilename));
			
			ret = true;
		}catch (Exception e){
			throw new Exception("fetchRev: Unable to fetch revision: " + e.toString());
		}
		return ret;
	}
	/**
	 * Create a new ZooKeeper session.
	 * 
	 * @throws Exception
	 */
	public void createZKSession() throws Exception{
		logger.debug("createZKSession");
		try{
			if (this.ZKAddr == null || this.ZKWatcher == null)
				throw new Exception("Invalid address or watcher object for ZooKeeper connection");
			this.ZKSession = new ZooKeeper(this.ZKAddr, S3vn.ZOOKEEPER_TIMEOUT, this.ZKWatcher);
		}catch (Exception e){
			throw new Exception("Could not establish ZooKeeper session: " + e.toString());
		}
	}
	/**
	 * Restore an existing ZooKeeper session based upon session ID in local session file.
	 * 
	 * @throws Exception
	 */
	public void restoreZKSession() throws Exception{
		logger.debug("restoreZKSession");
		byte pw[] = new byte[1];
		long sessionID = -1;
		String path = _session_filename();
		try{
			sessionID = (long)_get_integer(path);
			logger.debug("restoreZKSession: restoring sessionID: " + sessionID);
			this.ZKSession = new ZooKeeper(this.ZKAddr, S3vn.ZOOKEEPER_TIMEOUT, this.ZKWatcher, sessionID, pw);
		}catch (Exception e){
			throw new Exception(String.format("[Error]: Could not restore ZooKeeper session from file: %s: %s", path, e.toString()));
		}
	}
	/**
	 * Save current ZooKeeper session in local session ID file
	 * 
	 * @throws Exception
	 */
	public void saveZKSession() throws Exception{
		logger.debug("saveZKSession");
		String path = null;
		long sessionID = 0;
		try{
			path = _session_filename();
			sessionID = this.ZKSession.getSessionId();
			File fd = new File(path);
			BufferedWriter bwr = new BufferedWriter(new FileWriter(fd));
			bwr.write(String.valueOf(sessionID));
			bwr.close();
		}catch (IOException e){
			throw new Exception("Could not save sessionID into file: " + path);
		}
	}
	/**
	 * End ZooKeeper session. Closes ZooKeeper handle.
	 * 
	 */
	public void endZKSession(){
		logger.debug("endZKSession");
		try{
			this.ZKSession.close();
		}catch (Exception e){
		}
	}
	/**
	 * Delete local ZooKeeper session ID file.
	 * 
	 * @throws Exception
	 */
	public void deleteZKSession() throws Exception{
		logger.debug("deleteZKSession");
		String path = _session_filename();
		File fd = new File(path);
		try{
			if (!fd.delete()) throw new Exception();
		}catch (Exception e){
			throw new Exception("Could not delete sessionID file: " + path);
		}
	}
	/**
	 * Delete local lock file and delete lock file from ZooKeeper.
	 * 
	 * @throws Exception
	 */
	public void releaseLock() throws Exception{
		logger.debug("releaseLock");
		long lockValue = _load_lock();
		String lockFmtString = "%s" + S3vn.LOCK_FORMAT_STR;
		String lockPath = String.format(lockFmtString, _lock_path(), lockValue);
		_delete_lock();
		this.ZKSession.delete(lockPath, -1);
	}
	/**
	 * Fetch current revision number from ZooKeeper. 
	 * 
	 * @return Current revision number.
	 * @throws Exception
	 */
	public int fetchCurrent() throws Exception{
		logger.debug("fetchCurrent");
		int current;
		String path;
		try{
			path = _zk_current_path();
			current = _fetch_int(path);
		}catch (Exception e){
			throw new Exception("Could not fetch current revision number from ZooKeeper: " + e.toString());
		}
		return current;
	}
	/**
	 * Watch for repository revision update. Upon repository update of revision greater than @currentRev,
	 * find latest revision.
	 * 
	 * @param currentRev Current known revision number
	 * @return Latest revision
	 * @throws Exception
	 */
	public synchronized long watchCurrent(long currentRev) throws Exception{
		logger.debug("watchCurrent");
		String currentPath = _zk_current_path();
		byte[] data = null;
		long latestRev = 0;
		long ret = -1;
		BigInteger bi;
		
		// Only break out of the loop if there's an exception or if there's a rev number greater than current.
		while (true){
			try{
				this.ZKSession.getData(currentPath, this.ZKWatcher, this.dataCallback, null);
				synchronized(this.ZKWatcher){
					System.out.println("watchCurrent: waiting1.");
					this.ZKWatcher.wait();// Wait until callback is fired, at which time this.dataCallback.notify() will be called. Then we can proceed.
					this.ZKSession.exists(currentPath, false);
				}
				data = this.ZKSession.getData(currentPath, false, null);
				bi = new BigInteger(data);
				latestRev = bi.longValue();
				System.out.println(String.format("watchCurrent: data received latestRev=%d", latestRev));
				if (latestRev > currentRev){
					ret = latestRev;
					break;
				}
			}catch (NumberFormatException e){
				throw new Exception("Received invalid latest rev: " + data);
			}
		}
		return ret; // Should never be -1
	}
	/**
	 * Set current revision in local file.
	 * 
	 * @param current Revision number to store.
	 * @return True or False depending upon success of storage of the local file.
	 * @throws Exception
	 */
	public boolean setCurrent(long current) throws Exception{
		logger.debug("setCurrent");
		String path = this.localRepo + "/db/current";
		boolean ret;
		try{
			_set_int(path, current);
			ret = true;
		} catch (Exception e){
			throw new Exception("Unable to set int: " + e.toString());
		}
		return ret;
	}
	private String getUUID() throws Exception{
		logger.debug("getUUID");
		String path = _uuid_filename();
		String uuid = null;
        File fd = new File(path);
        BufferedReader bfd = new BufferedReader(new FileReader(fd));
        if ((uuid = bfd.readLine()) == null){
        	throw new Exception("Could not read UUID: " + path);
        }
        return uuid;
	}
	/**
	 * Store current revision number in ZooKeeper.
	 * 
	 * @param revNum Revision number to store.
	 * @throws Exception
	 */
	public void storeCurrent(long revNum) throws Exception{
		logger.debug("storeCurrent");
		String path = _zk_current_path();
		System.out.println("storeCurrent: " + path);
		_store_int(path, revNum);
		System.out.println("storeCurrent: stored revNum: " + revNum);
	}
	/**
	 * Get current revision number from local file.
	 * 
	 * @return Current revision number.
	 * @throws Exception
	 */
	public long getCurrent() throws Exception{
		logger.debug("getCurrent");
		String currentFile = _get_current_file_name();
		return _get_integer(currentFile);
	}
	/**
	 * Set program name to distinguish between processes or threads.
	 * 
	 * @param progName
	 */
	public void setProgName(String progName){
		this.progName = progName;
	}
	private static long _get_integer(String path) throws Exception{
		long integer;
		try{
			File fd = new File(path);
			BufferedReader bfd = new BufferedReader(new FileReader(fd));
			String data = null;
			if ((data = bfd.readLine()) != null){
				integer = Long.parseLong(data);
			}else{
				throw new Exception("Did not read an integer from file: " + path);
			}   
		}catch (FileNotFoundException e){ 
			throw new Exception("File does not exist: " + path);
		}catch (Exception e){ 
			throw new Exception(e.toString());
		}   
		return integer;
	}
	/**
	 * Convert integer portion of String @path to an integer.
	 * 
	 * @param path Path to search for integer.
	 * @return
	 */
	public int _convert_path_to_int(String path){
		String tmp[] = path.split("-");
		if (tmp.length > 0){
			return Integer.parseInt(tmp[tmp.length - 1]);
		}
		return -1;
	}
	private void _save_lock(int lockValue) throws Exception{
		try{
			String path = _lock_local_filename();
			File fd = new File(path);
			BufferedWriter bwr = new BufferedWriter(new FileWriter(fd));
			bwr.write(String.valueOf(lockValue));
			bwr.close();
		}catch (IOException e){
			throw new Exception("Could not save lock value to the local repository: " + lockValue);
		}
	}
	private long _load_lock() throws Exception{
		String path = _lock_local_filename();
		long lockVal = _get_integer(path);
		return lockVal;
	}
	private void _delete_lock() throws Exception{
		String path = _lock_local_filename();
		File fd = null;
		try{
			path = _lock_local_filename();
			fd = new File(path);
			if (!fd.delete()) throw new Exception();
		}catch (Exception e){
			throw new Exception("Could not delete local lock file: " + path);
		}
	}
	private String _session_filename(){
		return String.format("%s/zookeeper_session", this.localRepo);
	}
	private String _lock_local_filename(){
		return String.format("%s/zk_lock", this.localRepo);
	}
	private String _lock_path(){
		return String.format("%s/lock-", _lock_node());
	}
	private String _lock_node(){
	   return String.format("%s/lock", this.ZKPrefix);
	}
	private String _get_current_file_name(){
		return String.format("%s/db/current", this.localRepo);
	}
	private String _revprop_shard_dir(long revNum){
		return String.format("%s/db/revprops/%d", this.localRepo, revNum/S3vn.SHARD_COUNT);
	}
	private String _rev_shard_dir(long revNum){
		return String.format("%s/db/revs/%d", this.localRepo, revNum/S3vn.SHARD_COUNT);
	}
	private String _revprop_filename(long revNum){
		long shard = revNum / S3vn.SHARD_COUNT;
		return String.format("%s/db/revprops/%d/%d", this.localRepo, shard, revNum);
	}
	private String _revprop_filename_tmp(long revNum){
		long shard = revNum / S3vn.SHARD_COUNT;
		return String.format("%s/db/revprops/%d/%d.%s", this.localRepo, shard, revNum, this.progName);
	}
	private String _rev_filename(long revNum){
		long shard = revNum / S3vn.SHARD_COUNT;
		return String.format("%s/db/revs/%d/%d", this.localRepo, shard, revNum);
	}
	private String _rev_filename_tmp(long revNum){
		long shard = revNum / S3vn.SHARD_COUNT;
		return String.format("%s/db/revs/%d/%d.%s", this.localRepo, shard, revNum, this.progName);
	}
	private String _rev_key(long revNum){
		return String.format("%srev-%d", this.S3Prefix, revNum);
	}
	private String _revprop_key(long revNum){
		return String.format("%srevprop-%d", this.S3Prefix, revNum);
	}
	private void _store_int(String path, long value) throws Exception{
		BigInteger bi = new BigInteger(String.valueOf(value));
		byte data[] = bi.toByteArray();
		System.out.println(String.format("_store_int: .%d:%s:%s.", value, String.valueOf(value),data));
		try {
			try{ /* Set the value */
				System.out.println(String.format("_store_int: Setting data in ZK: path=%s data=%s", path, data));
				this.ZKSession.setData(path, data, -1);
			}catch (KeeperException.NoNodeException e){ // If the node did not exist, create the node with the value
				this.ZKSession.create(path, data, Ids.OPEN_ACL_UNSAFE, CreateMode.fromFlag(0));
			}
		}catch (Exception e){
			throw new Exception("Could not store value in ZooKeeper");
		}
	}
	private void _set_int(String path, long value) throws IOException{
		String tmp = String.format("%s/db/current.%s", this.localRepo, this.progName); // This should perhaps be randomly-generated
		FileWriter tmpFd = null;
		File tmpFile = null;
		tmpFd = new FileWriter(tmp);
		tmpFd.write(String.valueOf(value) + "\n");
		tmpFd.close();
		tmpFile = new File(tmp);
		tmpFile.renameTo(new File(path));
		return;
	}
	private int _fetch_int(String path) throws Exception{
		int ret;
		BigInteger bi;
		byte data[] = new byte[1024];
		if (this.ZKSession == null)
			this.createZKSession();
		data = this.ZKSession.getData(path, null, null);
		bi = new BigInteger(data);
		ret = bi.intValue();
		return ret;
	}
	private String _zk_current_path(){
		return String.format("%s/current", this.ZKPrefix);
	}
	private String _uuid_filename(){
		return String.format("%s/db/uuid", this.localRepo);
	}
	private String _uuid_key(){
		//return String.format("%suuid", this.S3Prefix);
		return "uuid";
	}
}
